package com.zwh.rocketmq.demo.normal;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * @author zwh
 * @date 2024-1-18 15:47
 */
@Service
public class NormalMessageProduct {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通消息
     */
    public void sendMsg() {
        String destination = "TEST_TOPIC";
        String tags = "tag1";
        String msgBody = "hello rocketMQ";
        String keys = "orderId1";
        if (StringUtils.isNotBlank(tags)) {
            // 带tag的消息，直接在topic后面加上":tag"， 底层会进行自动转换
            destination = destination + ":" + tags;
        }
        /*
         headers中的key：
            id: 表示消息的id,和mq控制台上的messageId不一样，没什么作用
            TAGS：消息标签
            KEYS： 表示消息的key，用于哈希索引，查找使用
         */
        Message<String> message = MessageBuilder.withPayload(msgBody)
                .setHeader(RocketMQHeaders.KEYS, keys)
                .build();

        // 同步发送
        // 同步发送方式请务必捕获发送异常，并做业务侧失败兜底逻辑，如果忽略异常则可能会导致消息未成功发送的情况。
        rocketMQTemplate.syncSend(destination, message);

        // 异步发送消息，有回调
//        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                // 处理消息发送成功逻辑
//            }
//            @Override
//            public void onException(Throwable e) {
//                // 处理消息发送异常逻辑
//            }
//        });

        // 发送一次，不管有没有ack响应，类似udp协议,
        // 典型应用于某些耗时非常短，但对可靠性要求并不高的场景，如 日志收集
//        rocketMQTemplate.sendOneWay(destination, message);
    }
}
